package org.gluedom.event;

import static com.google.common.base.Preconditions.checkNotNull;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;

import org.gluedom.data.Identifier;
import org.gluedom.data.Immutable;
import org.gluedom.transport.Connection;
import org.gluedom.transport.ConnectionCallable;
import org.gluedom.transport.ReceiveCallback;
import org.gluedom.transport.SendCallback;
import org.gluedom.transport.Transports;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Multimap;

public class DefaultEventBus implements EventBus {
	private final Multimap<Identifier, Connection> connectionMap = ArrayListMultimap
			.create();
	private ExecutorService executor = Executors.newCachedThreadPool();
	private AtomicBoolean started = new AtomicBoolean(false);
	private Connection connection = null;

	@Override
	public void register(Connection connection,
			ImmutableList<Identifier> methods) {
		if (started.get())
			throw new IllegalStateException("this eventbus was already started");
		checkNotNull(connection);
		for (Identifier method : methods)
			connectionMap.put(method, connection);
	}

	@Override
	public void start(Connection connection) {
		if (started.compareAndSet(false, true)) {
			this.connection = connection;
			Transports.getTransport(connection).receive(new EventBusCallback());
		} else
			throw new IllegalStateException("this eventbus was already started");
	}

	@Override
	public void stop() {
		if (started.compareAndSet(true, false)) {
			Transports.getTransport(connection).receive(null);
		} else
			throw new IllegalStateException("this eventbus was already stopped");
	}

	private class EventBusCallback implements ReceiveCallback {
		@Override
		public void doCallback(ImmutableList<Immutable> data,
				final SendCallback callback) {
			Event<Immutable> event = new Event.Builder<Immutable>(data)
					.build();
			ImmutableList<Immutable> result = ImmutableList.of();
			if (event.isConsumable()) {
				for (final Connection con : connectionMap.get(event.getType())) {
					result = Transports.sendSync(con, data);
					if (result != null && result.size() > 0)
						break;
				}
			} else {
				List<Callable<ImmutableList<Immutable>>> taskList = new ArrayList<Callable<ImmutableList<Immutable>>>();
				for (final Connection con : connectionMap.get(event.getType()))
					taskList.add(new ConnectionCallable(con, data));
				List<Future<ImmutableList<Immutable>>> resultList = null;
				try {
					resultList = executor
							.invokeAll(taskList, Transports.DEFAULT_TIMEOUT,
									Transports.DEFAULT_UNIT);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				if (resultList != null)
					for (Future<ImmutableList<Immutable>> list : resultList)
						try {
							if (!(list.isCancelled()) && list.get().size() > 0)
								result = list.get();
						} catch (InterruptedException e) {
							e.printStackTrace();
						} catch (ExecutionException e) {
							e.printStackTrace();
						}
			}
			final ImmutableList<Immutable> finalResult = result;
			executor.execute(new Runnable() {
				@Override
				public void run() {
					callback.doCallback(finalResult);
				}
			});
		}
	}
}